package cgl.hadoopsensorgrid.sensorgrid;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.util.GenericOptionsParser;

import cgl.hadoopsensorgrid.sensorgrid.rtdtohadoop.rtdController;
import cgl.hadoopsensorgrid.sensorgrid.ryotoeval.ryoController;
import cgl.hadoopsensorgrid.sensorgrid.eval.evalController;

/***
 * 
 * The central controller takes arguments from command line; it then uses them
 * to create controllers for each step of the data processing and deployes the
 * controllers. Each subcontroller is started in a nested class which extends
 * class Thread. RYO and EVAL controllers run serially. They run concurrently
 * with RTD controller and are started and stopped periodically. RTD controller
 * runs in a separate thread.
 * 
 * @author Chao Sun
 * 
 */

public class mainController {
	private String rtdNetworkList;
	private String rtdOutPath;
	private String ryoSourceList;
	private int ryoReducerNum;
	private String ryoOutPath;
	private String evalDataList;
	private int evalReducerNum;
	private String evalOutPath;

	/***
	 * 
	 * Class for starting the rtd job. Runs in a separate thread.
	 * 
	 * @author Chao Sun
	 * 
	 */
	class startRtd extends Thread {

		/**
		 * Override the run() method. Deploy the rtdController.
		 */
		public void run() {
			try {
				rtdController rtd = new rtdController(rtdNetworkList, rtdOutPath);
				rtd.deploy();
			} catch (Exception rtdE) {
				System.err.println("Rtd controller error.");
				rtdE.printStackTrace();
			}
		}

	}

	/**
	 * Constructor. Reads the arguments to class variables.
	 * 
	 * @param args
	 *          command line string array, received from the main() method
	 */
	public mainController(String args[]) {
		rtdNetworkList = args[0];
		rtdOutPath = args[1];
		ryoSourceList = args[2];
		ryoReducerNum = Integer.parseInt(args[3]);
		ryoOutPath = args[4];
		evalDataList = args[5];
		evalReducerNum = Integer.parseInt(args[6]);
		evalOutPath = args[7];
	}

	/**
	 * Start all the jobs.
	 * 
	 * @throws InterruptedException
	 */
	public void startJobs() throws InterruptedException {
		// start the rtd job in another thread
		startRtd rtd = new startRtd();
		rtd.start();
		Thread.sleep(1000);
		// periodically start the ryo and eval jobs
		while (true) {
			try {
				ryoController ryo = new ryoController(ryoSourceList, ryoReducerNum,
						ryoOutPath);
				ryo.deploy();
			} catch (Exception ryoE) {
				System.err.println("Ryo controller error.");
				ryoE.printStackTrace();
			}

			Thread.sleep(100);

			try {
				evalController eval = new evalController(evalDataList, evalReducerNum,
						evalOutPath);
				eval.deploy();
			} catch (Exception evalE) {
				System.err.println("Eval controller error.");
				evalE.printStackTrace();
			}
		}
	}

	/**
	 * main method.
	 * 
	 * @param args
	 *          command line string array
	 */
	public static void main(String args[]) {
		if (args.length < 8) {
			System.err.println("Arguments missing. Usage: "
					+ "<rtdNetworkList> <rtdOutPath> "
					+ "<ryoSourceList> <ryoReducerNum> <ryoOutPath> "
					+ "<evalDataList> <evalReducerNum> <evalOutPath>");
			System.exit(1);
		}

		try {
			mainController controller = new mainController(args);
			controller.startJobs();
		} catch (InterruptedException ie) {
			System.err.println("Jobs error");
		}
	}

}